package com.wzh.project.kafka.provider;

import java.util.Properties;
import java.util.Random;
import java.util.concurrent.Future;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import com.wzh.project.common.Constants;

/**
 * 用于生产消息
 * @author wzh
 *
 */
public class KafkaProviderThread implements Runnable{
	
	private static final Logger logger=LogManager.getLogger(KafkaProviderThread.class);
	
	@Override
	public void run() {
		
	}
	public static void main(String[] args) throws InterruptedException {
		sendMessage("");
	}
	
	public static void sendMessage(String mess) throws InterruptedException { 
		Properties p = new Properties();
	    p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.23.11.132:9092");//kafka地址，多个地址用逗号分割
	    p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
	    p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
	    p.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
	    KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(p);
	
	    try {
	        while (true) {
	            String msg = "您好," + new Random().nextInt(100);
//	            new ProducerRecord<String,Object>();
	            ProducerRecord<String, String> record = new ProducerRecord<String, String>(Constants.KAFKA_TOPIC, msg);
	            Future<RecordMetadata> result=kafkaProducer.send(record,new Callback() {
                    public void onCompletion(RecordMetadata metadata, Exception e) {
                        if(e != null) {
                           e.printStackTrace();
                        } else {
                        	logger.debug("The offset of the record we just sent is: " + metadata.offset());
                        }
                    }
                });
	            logger.debug("消息发送成功:"+msg);
	            Thread.sleep(1500);
	        }
	    } finally {
	        kafkaProducer.close();
	    }
	 }

}
